// com/tidu/disruptor/handler/WriteDataEventHandler.java
package com.tidu.disruptor.handler;

import com.lmax.disruptor.WorkHandler;
import com.tidu.disruptor.event.DataSyncEvent;
import com.tidu.disruptor.manager.BatchAckManager;
import com.tidu.strategy.sync.SyncStrategy;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ErrorHandler implements WorkHandler<DataSyncEvent> {

    private final SyncStrategy syncStrategy;
    private final BatchAckManager batchAckManager;
    private final String handlerId;

    public ErrorHandler(SyncStrategy syncStrategy, BatchAckManager batchAckManager, String handlerId) {
        this.syncStrategy = syncStrategy;
        this.batchAckManager = batchAckManager;
        this.handlerId = handlerId;
    }

    @Override
    public void onEvent(DataSyncEvent event) {
        boolean success = false;
        String failReason = null;

        try {
            log.info("Handler {} processing batch {} with {} records",
                    handlerId, event.getBatchId(), event.getData().size());

            if(0/1 == 0) {
                throw new RuntimeException("故意的");
            }
            syncStrategy.writeData(event.getData());

            System.out.println("=====================================================================================================");
            System.out.println("当前进度: \n" + syncStrategy.getProgress());
            System.out.println("=====================================================================================================");

            success = true;
        } catch (Exception e) {
            failReason = e.getMessage();
            log.info("Handler {} failed to process batch {} of size {}", handlerId, event.getBatchId(), event.getData().size());
            log.error("Handler {} failed to process batch {} of size {}: {}",
                    handlerId, event.getBatchId(), event.getData().size(), e.getMessage(), e);
        } finally {
            // 无论成功失败都确认批次处理完成
            if (batchAckManager != null) {
                batchAckManager.acknowledgeBatch(event.getBatchId(), success, failReason);
            }
        }
    }
}
